/**同步、异步、单向、事务、请求响应消息发送
 * /rocketmq/convert
 * /rocketmq/sync/{msg}
 * /rocketmq/sync/user
 * /rocketmq/sync/batch/{msg}/{batchSize}
 * /rocketmq/sync/tag/{msg}
 * /rocketmq/sync/orderly
 * /rocketmq/sync/reply/{msg}
 *
 * /rocketmq/async/user
 * /rocketmq/async/user/delay
 * /rocketmq/async/orderly
 * /rocketmq/async/reply
 *
 * /rocketmq/oneway/{msg}
 * /rocketmq/oneway/orderly
 *
 * /rocketmq/trans
 */
package com.example.rocketmqspringbootdemo.controller;

import com.example.rocketmqspringbootdemo.CommonConstrant;
import com.example.rocketmqspringbootdemo.service.producer.SendService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import com.example.rocketmqspringbootdemo.pojo.User;

import javax.annotation.Resource;

@Slf4j
@RestController
@RequestMapping("/rocketmq")
public class RocketMQController {
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    private final SendService sendService;
    @Autowired
    public RocketMQController(SendService sendService) {
        this.sendService = sendService;
    }

    //同步发送，相当于covert+syncSend，(Convet：object+header+postprcessor-->Message<?>)
    @RequestMapping("/convert")
    public String convertSend() {
        this.rocketMQTemplate.convertAndSend(CommonConstrant.RocketMQ.STRING_TOPIC, "This is from convertAndSend......");
        User user = new User("zhaoliu", 50);

        this.rocketMQTemplate.convertAndSend(CommonConstrant.RocketMQ.OBJECT_TOPIC + ":Tag0", user);
       /*重载方法：
           convertAndSend(Object payload)
           convertAndSend(Object payload, @Nullable MessagePostProcessor postProcessor)
           convertAndSend(D destination, Object payload, Map<String, Object> headers)
           convertAndSend(D destination, Object payload, MessagePostProcessor postProcessor)
           convertAndSend(D destination, Object payload, @Nullable Map<String, Object> headers,@Nullable MessagePostProcessor postProcessor)
        */
        return "convertAndSend .......";
    }


    /*---------------------------------同步发送------------------------------*/
    @GetMapping(value = "/sync/{msg}")
    public String sendString(@PathVariable String msg) {
        try {
            return this.sendService.syncSendStringMessage(msg, CommonConstrant.RocketMQ.STRING_TOPIC);
        } catch (IllegalArgumentException e) {
            log.warn("sendString -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("sendString -- error --> ", e);
            return "Error:发送失败!";
        }
    }

    @GetMapping(value = "/sync/user")
    public String sendUser() {
        try {
            User user = new User("zhangsan", 20);
            return this.sendService.syncSendObjectMessage(user, CommonConstrant.RocketMQ.OBJECT_TOPIC);
        } catch (IllegalArgumentException e) {
            log.warn("sendUser -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("sendUser -- error --> ", e);
            return "Error:发送失败!";
        }
    }

    @GetMapping(value = "/sync/batch/{msg}/{batchSize}")
    public String sendCollectionWithTag(@PathVariable(value="msg") String msg, @PathVariable(value="batchSize") Integer batchSize) {
        try {
             this.sendService.syncSendCollectionMessage(msg, CommonConstrant.RocketMQ.COLLECTION_TOPIC,
                    CommonConstrant.RocketMQ.TAG_BATCH, batchSize);
        } catch (IllegalArgumentException e) {
            log.warn("sendStringWithTag -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("sendStringWithTag -- error --> ", e);
            return "Error:发送失败!";
        }
        return "Sync send collection.......";
    }

    @GetMapping(value = "/sync/tag/{msg}", produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendStringWithTag(@PathVariable String msg) {
        try {
            return this.sendService.syncSendStringMessageWithTag(msg, CommonConstrant.RocketMQ.STRING_TOPIC,
                    CommonConstrant.RocketMQ.TAG_STRING);
        } catch (IllegalArgumentException e) {
            log.warn("sendStringWithTag -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("sendStringWithTag -- error --> ", e);
            return "Error:发送失败!";
        }
    }

    @GetMapping(value = "/sync/orderly", produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendStringOrderly() {
        try {
            return this.sendService.syncSendBatchStringMessageOrderly(CommonConstrant.RocketMQ.STRING_TOPIC_ORDERLY);
        } catch (IllegalArgumentException e) {
            log.warn("sendStringOrderly -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("sendStringOrderly -- error --> ", e);
            return "Error:发送失败!";
        }
    }

    @GetMapping(value = "/sync/reply/{msg}", produces = MediaType.APPLICATION_JSON_VALUE)
    public String replySendMessage(@PathVariable(value = "msg") String msg) {
        try {
            return sendService.syncSendStringReply(msg, CommonConstrant.RocketMQ.STRING_TOPIC_REPLY, "TagString");
        } catch (IllegalArgumentException e) {
            log.warn("replySendMessage -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("replySendMessage -- error --> ", e);
            return "Error:发送失败!";
        }
    }

    /*---------------------------------异步发送------------------------------*/
    @GetMapping(value = "/async/user", produces = MediaType.APPLICATION_JSON_VALUE)
    public String asyncSendUser() {
        try {
            User user = new User("lisi", 30);
            return this.sendService.asyncSendObjectMessage(user, CommonConstrant.RocketMQ.OBJECT_TOPIC);
        } catch (IllegalArgumentException e) {
            log.warn("asyncSendUser -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("asyncSendUser -- error --> ", e);
            return "Error:发送失败!";
        }
    }

    @GetMapping(value = "/async/user/delay", produces = MediaType.APPLICATION_JSON_VALUE)
    public String asyncSendUserWithTagDelay() {
        try {
            User user = new User("wangwu", 40);
            return this.sendService.asyncSendDelayWithTag(user, CommonConstrant.RocketMQ.OBJECT_TOPIC,
                    CommonConstrant.RocketMQ.TAG_OBJECT, 3);
        } catch (IllegalArgumentException e) {
            log.warn("sendUserWithTagDelay -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("sendUserWithTagDelay -- error --> ", e);
            return "Error:发送失败!";
        }
    }

    @GetMapping(value = "/async/orderly", produces = MediaType.APPLICATION_JSON_VALUE)
    public String asyncSendMessageOrderly() {
        try {
            return this.sendService.asyncSendOrderly(CommonConstrant.RocketMQ.STRING_TOPIC_ORDERLY);
        } catch (IllegalArgumentException e) {
            log.warn("sendUserWithTagDelay -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("sendUserWithTagDelay -- error --> ", e);
            return "Error:发送失败!";
        }
    }

    // Send request in async mode and receive a reply of User type.
    @GetMapping(value = "/async/reply", produces = MediaType.APPLICATION_JSON_VALUE)
    public String asyncSendUserReply() {
        String result;
        try {
             User user = new User("zhaoliu", 60);
             result=this.sendService.asyncSendUserReply(user,CommonConstrant.RocketMQ.STRING_TOPIC_REPLY,"TagUser");
        } catch (IllegalArgumentException e) {
            log.warn("sendUserWithTagDelay -- param fail --> {}", e.getMessage());
            result="Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("sendUserWithTagDelay -- error --> ", e);
            result= "Error:发送失败!";
        }

        return result;
    }

    /*---------------------------------单向发送------------------------------*/
    @GetMapping(value = "/oneway/{msg}", produces = MediaType.APPLICATION_JSON_VALUE)
    public String onewaySendMessage(@PathVariable(value = "msg") String msg) {
        try {
            sendService.onewaySendString(msg, CommonConstrant.RocketMQ.STRING_TOPIC);
        } catch (IllegalArgumentException e) {
            log.warn("onewaySendMessage -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("onewaySendMessage -- error --> ", e);
            return "Error:发送失败!";
        }
        return "Oneway send ......!";
    }

    @GetMapping(value = "/oneway/orderly", produces = MediaType.APPLICATION_JSON_VALUE)
    public String onewaySendMessageOrderly() {
        try {
            sendService.onewaySendStringOrderly(CommonConstrant.RocketMQ.STRING_TOPIC_ORDERLY);
        } catch (IllegalArgumentException e) {
            log.warn("onewaySendMessageOrderly -- param fail --> {}", e.getMessage());
            return "Error:IllegalArgument!";
        } catch (Exception e) {
            log.warn("onewaySendMessageOrderly -- error --> ", e);
            return "Error:发送失败!";
        }
        return "Oneway send orderly ......";
    }


    @GetMapping(value = "/trans")
    public String transactionSend() {
        sendService.sendTransactionMessage(CommonConstrant.RocketMQ.TRANSACTION_TOPIC);
        return "Transction Send......";
    }
}